package com.lagou.zk;

import com.alibaba.fastjson.JSON;
import com.lagou.zk.entity.MysqlConfig;
import com.mchange.v2.c3p0.ComboPooledDataSource;
import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.exception.ZkMarshallingError;
import org.I0Itec.zkclient.serialize.ZkSerializer;

import java.nio.charset.StandardCharsets;

public class MysqlPool {

    private static ComboPooledDataSource pools = null;

    private ZkClient zkClient = null;

    public static final String MYSQL_CONFIG_PATH = "/mysql/config";

    public MysqlPool() {

        synchronized (MysqlPool.class) {
            if (null == pools) {
                this.zkClient = new ZkClient("centos1:2181,centos2:2181,centos3:2181");
                this.zkClient.setZkSerializer(new ZkSerializer() {
                    @Override
                    public byte[] serialize(Object o) throws ZkMarshallingError {
                        return String.valueOf(o).getBytes(StandardCharsets.UTF_8);
                    }

                    @Override
                    public Object deserialize(byte[] bytes) throws ZkMarshallingError {
                        return new String(bytes, StandardCharsets.UTF_8);
                    }
                });
                System.out.println("zookeeper 初始化成功");
                zkWatch();
                initMysqlPool();
                System.out.println("mysql pool 初始化成功");
            }
        }
    }

    private void zkWatch() {
        if (!zkClient.exists(MYSQL_CONFIG_PATH)) {
            zkClient.createPersistent(MYSQL_CONFIG_PATH, true);
            zkClient.writeData(MYSQL_CONFIG_PATH, "{\"dbHost\":\"centos1:3306\",\"db\":\"azkaban\",\"username\":\"azkaban\",\"password\":\"azkaban\"}");
        }
        zkClient.subscribeDataChanges(MYSQL_CONFIG_PATH, new IZkDataListener() {
            @Override
            public void handleDataChange(String s, Object o) throws Exception {
                System.out.println("监听到mysql配置修改->s" + o);
                closeOldPool();
                initMysqlPool();
            }

            @Override
            public void handleDataDeleted(String s) throws Exception {

            }
        });
        System.out.println("zkWatch 启动成功");
    }


    private void initMysqlPool() {
        pools = new ComboPooledDataSource();
        try {
            pools.setDriverClass("com.mysql.jdbc.Driver");
        } catch (Exception e) {
            e.printStackTrace();
        }

        MysqlConfig mysqlConfig = getMysqlConfig();
        System.out.println("获取到mysql配置->" + JSON.toJSONString(mysqlConfig));
        pools.setJdbcUrl(String.format("jdbc:mysql://%s/%s", mysqlConfig.getDbHost(), mysqlConfig.getDb()));
        pools.setUser(mysqlConfig.getUsername());
        pools.setPassword(mysqlConfig.getPassword());
        pools.setMinPoolSize(5);
        pools.setMaxPoolSize(20);
        pools.setAcquireIncrement(5);
    }

    private MysqlConfig getMysqlConfig() {
        Object o = zkClient.readData("/mysql/config");
        return JSON.parseObject(String.valueOf(o), MysqlConfig.class);
    }

    private void closeOldPool() {
        if (null != pools) {
            pools.close();
            System.out.println("关闭mysql连接池完成");
        }
    }

    public ComboPooledDataSource getPools() {
        return pools;
    }
}
